package org.piggy.core.disruptor;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WorkerPool;
import org.piggy.core.netty.wrapper.HttpRequestWrapper;
import org.piggy.core.netty.processor.DefaultNettyProcessor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

public class DisruptorQueueHandler {

    private final WorkerPool<DisruptorEvent> workerPool;
    private final RingBuffer<DisruptorEvent> ringBuffer;
    private final ExecutorService executorService;
    private final DisruptorEventTranslatorOneArg eventTranslatorOneArg;

    public DisruptorQueueHandler(DisruptorQueueHandler.Builder builder, DefaultNettyProcessor defaultNettyProcessor) {
        this.eventTranslatorOneArg = new DisruptorEventTranslatorOneArg();

        DisruptorWorkHandler[] workHandlers = new DisruptorWorkHandler[builder.threads];
        for (int i = 0; i < builder.threads; i++) {
            workHandlers[i] = new DisruptorWorkHandler(defaultNettyProcessor);
        }
        DisruptorEventFactory piggyEventFactory = new DisruptorEventFactory();
        DisruptorExceptionHandler piggyExceptionHandler = new DisruptorExceptionHandler();
        this.workerPool = new WorkerPool<>(piggyEventFactory, piggyExceptionHandler, workHandlers);
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("piggy-queue-handler-pool-%d").build();
        this.executorService = Executors.newFixedThreadPool(builder.threads, threadFactory);
        this.ringBuffer = this.workerPool.start(executorService);
    }

    public void add(HttpRequestWrapper wrapper) {
        this.ringBuffer.publishEvent(eventTranslatorOneArg, wrapper);
    }

    public void shutDown(){
        // 等待RingBuffer中已发布的事件全部处理完毕，然后停止工作线程
        this.workerPool.drainAndHalt();
        // 关闭
        this.executorService.shutdown();
    }

    public static class Builder {
        private int threads = 1;

        public DisruptorQueueHandler.Builder setThreads(int threads) {
            if(threads > 0){
                this.threads = threads;
            }
            return this;
        }
    }

}
